package com.example.rocket.producer;

import com.example.rocket.config.NbMqProperties;
import com.example.rocket.config.NbProducerProperties;
import com.example.rocket.core.NbCallback;
import com.example.rocket.core.NbSendResult;
import com.example.rocket.utils.JSONUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Objects;


/**
 * @author TAO
 * @description: RockerMQProducer - 采用 rocker-client
 * @date 2021/12/20 7:25 下午
 */
@Slf4j
public class RocketMqProducer implements NbMqProducer {

    private static DefaultMQProducer producer;

    public RocketMqProducer(NbMqProperties properties) throws MQClientException {
        log.info("↓↓↓↓----↓↓↓↓----↓↓↓↓----↓↓↓↓");
        log.warn("-->当前环境启用===>RocketMQ<--");
        log.info("↑↑↑↑----↑↑↑↑----↑↑↑↑----↑↑↑↑");
        NbProducerProperties producerProperties = properties.getProducer();

        producer = new DefaultMQProducer(producerProperties.getProducerId());

        producer.setNamesrvAddr(properties.getNameServer());

        producer.start();
    }


    /**
     * 同步消息
     *
     * @param topic
     * @param tag
     * @param payload
     * @return
     */
    @Override
    public String syncSend(String topic, String tag, Object payload) {
        try {
            Message msg = new Message(topic, tag, JSONUtils.writeAsBytes(payload));
            SendResult sendResult = producer.send(msg);
            return sendResult.getMsgId();
        } catch (Exception e) {
            log.error("发送消息异常", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 异步消息
     *
     * @param topic
     * @param tag
     * @param payload
     * @param sendCallback
     */
    @Override
    public void asyncSend(String topic, String tag, Object payload, NbCallback sendCallback) {
        try {
            Message msg = new Message(topic, tag, JSONUtils.writeAsBytes(payload));
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.trace("内部asyncSend调用成功返回===>{}", sendResult);
                    sendCallback.onSuccess(sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    log.error("内部asyncSend调用失败异常返回===>{}", e);
                    sendCallback.onException(e.getCause());
                }
            });
        } catch (Exception e) {
            log.error("发送消息异常", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 单向消息
     *
     * @param topic
     * @param tag
     * @param payload
     */
    @Override
    public void sendOneWay(String topic, String tag, Object payload) {
        try {
            Message msg = new Message(topic, tag, JSONUtils.writeAsBytes(payload));
            producer.sendOneway(msg);
        } catch (Exception e) {
            log.error("发送消息异常", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 批量消息-同步
     *
     * @param topic
     * @param tag
     * @param messages
     * @return
     */
    @Override
    public NbSendResult batchSend(String topic, String tag, Collection<Object> messages) {
        if (Objects.isNull(messages) || messages.size() == 0) {
            log.error("批量同步发送失败。目的地：{}，消息为空", topic);
            throw new IllegalArgumentException("`messages` 不能为空");
        }
        try {
            long now = System.currentTimeMillis();
            Collection<Message> msgList = new ArrayList<>();
            for (Object msg : messages) {
                //if (Objects.isNull(msg) || Objects.isNull(msg.getPayload())) {
                if (Objects.isNull(msg)) {
                    log.warn("在批处理中发现一条消息为空，跳过它");
                    continue;
                }
                msgList.add(this.createRocketMqMessage(topic, tag, msg));
            }
            SendResult sendResult = producer.send(msgList);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("发送消息成本：{} 毫秒，msgId：{}", costTime, sendResult.getMsgId());
            }
            return new NbSendResult(sendResult);
        } catch (Exception e) {
            log.error("批量同步发送失败。目的地：{},消息大小：{}", topic, messages.size());
            throw new IllegalArgumentException("`messages` can not be empty");
        }
    }

    /**
     * 同步延迟消息
     *
     * @param topic
     * @param tag
     * @param message
     * @param timeout
     * @param delayLevel
     * @return
     */
    @Override
    public NbSendResult syncSend(String topic, String tag, Object message, long timeout, int delayLevel) {
        //if (Objects.isNull(message) || Objects.isNull(message.getPayload())){
        if (Objects.isNull(message)) {
            log.error("同步发送失败。目的地：{}，消息为空", topic);
            throw new IllegalArgumentException("`message` 和 `message.payload` 不能为空");
        }
        try {
            long now = System.currentTimeMillis();
            Message rocketMsg = this.createRocketMqMessage(topic, tag, message);
            if (delayLevel > 0) {
                rocketMsg.setDelayTimeLevel(delayLevel);
            }
            SendResult sendResult = producer.send(rocketMsg, timeout);
            long costTime = System.currentTimeMillis() - now;
            if (log.isDebugEnabled()) {
                log.debug("发送消息成本：{} 毫秒，msgId：{}", costTime, sendResult.getMsgId());
            }
            return new NbSendResult(sendResult);
        } catch (Exception e) {
            log.error("同步发送失败。目的地：{}，消息：{}} ", topic, message);
            throw new IllegalArgumentException("`message` 和 `message.payload` 不能为空");
        }
    }

    /**
     * 异步延迟消息
     *  @param topic
     * @param tag
     * @param message
     * @param timeout
     * @param delayLevel
     * @param sendCallback
     */
    @Override
    public void asyncSend(String topic, String tag, Object message, long timeout, int delayLevel, NbCallback sendCallback) {
        //if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
        if (Objects.isNull(message)) {
            log.error("异步发送失败。目的地：{}，消息为空", topic);
            throw new IllegalArgumentException("`message` 和 `message.payload` 不能为空");
        }
        try {
            Message rocketMsg = this.createRocketMqMessage(topic, tag, message);
            if (delayLevel > 0) {
                rocketMsg.setDelayTimeLevel(delayLevel);
            }
            producer.send(rocketMsg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.trace("内部asyncSend调用成功返回===>{}", sendResult);
                    sendCallback.onSuccess(sendResult.getMsgId());
                }
                @Override
                public void onException(Throwable e) {
                    log.error("内部asyncSend调用失败异常返回===>{}", e);
                    sendCallback.onException(e.getCause());
                }
            }, timeout);
        } catch (Exception e) {
            log.info("异步发送失败。目的地：{}，消息：{} ", topic, message);
            throw new RuntimeException(e);
        }
    }

    public Message createRocketMqMessage(String topic, String tag, Object msg) {
        return new Message(topic, tag, JSONUtils.writeAsBytes(msg));
    }

}
